package com.saga.energy.source;

import com.saga.energy.bean.User;
import com.saga.energy.util.SysConstant;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.*;

public class ZJMySQLSource extends RichSourceFunction<User> {

    private PreparedStatement ps1;
    private Connection conn1;

    @Override
    public void open(Configuration parameters) throws Exception {
        System.out.println("init conn1");
        conn1 = DriverManager.getConnection(SysConstant.ZJ_MYSQL_URL, SysConstant.ZJ_MYSQL_USER, SysConstant.ZJ_MYSQL_PASSWD);
        ps1 = conn1.prepareStatement("select id, `name`, company_id, `type`, `status`, create_date, update_date, dep_name, job, phone, manage_user_type, first_use_time from `user`;");
//        conn2 = DriverManager.getConnection("jdbc:mysql://localhost:3306/test?autoReconnect=true&useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8&useSSL=false", "root", "root");
//        ps2 = conn2.prepareStatement("select space_id, space_name, room_id, room_name, building_name, floor_name from room_to_space;");
    }

    @Override
    public void run(SourceContext<User> sourceContext) throws Exception {
        System.out.println("query");


        ResultSet resultSet = ps1.executeQuery();
        while (resultSet.next()) {
            User user = new User();
            user.setId(resultSet.getString("id"));
            user.setName(resultSet.getString("name"));
            user.setCompanyId(resultSet.getString("company_id"));
            user.setType(resultSet.getInt("type"));
            user.setStatus(resultSet.getInt("status"));
            user.setCreateDate(resultSet.getDate("create_date"));
            user.setUpdateDate(resultSet.getDate("update_date"));
            user.setDepName(resultSet.getString("dep_name"));
            user.setJob(resultSet.getString("job"));
            user.setPhone(resultSet.getString("phone"));
            user.setManageUserType(resultSet.getInt("manage_user_type"));
            user.setFirstUseTime(resultSet.getString("first_use_time"));
            sourceContext.collect(user);
        }
    }

    @Override
    public void close() throws Exception {
        ps1.close();
        conn1.close();
    }

    @Override
    public void cancel() {

    }
}
